package com.sugon.ohdfs.integration.flink.job.stream.source;

import com.google.common.base.Strings;
import com.sugon.ohdfs.integration.flink.domain.TestItem;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineFormat;
import org.apache.flink.core.fs.Path;
import org.springframework.beans.factory.annotation.Value;

import java.time.Duration;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Arrays;

public class RecordStreamSourceJob extends AbstractStreamSourceJob {

    @Value("${hdfs.base-path:hdfs://10.11.8.29:9001/flink/sinkTest/stream/sink/row}")
    String basePath;

    @Override
    protected FileSource buildFileSource() {
        return FileSource
                .forRecordStreamFormat(new TextLineFormat(), new Path(basePath))
                .monitorContinuously(Duration.ofMillis(5))
                .build();
    }

    /**
     * 反序列化数据，这里是直接反序列化的toString，所以比较复杂，应该写json的
     * @param item
     * @return
     */
    @Override
    protected TestItem transform(Object item) {
        String value = item.toString();

        if(Strings.isNullOrEmpty(value)){
            return null;
        }
        if (value.equals("TestItem")){
            return null;
        }

        TestItem result = new TestItem();
        Arrays.stream(value.substring(9, value.length() - 2).split(", "))
                .map(slice -> slice.split("="))
                .forEach(array -> {
                    if(array.length==0 || Strings.isNullOrEmpty(array[0])){
                        return;
                    }
                    switch (array[0]){
                        case "name":
                            result.setName(array[1]);
                            break;
                        case "value":
                            result.setValue(array[1]);
                            break;
                        case "produceTimestamp":
                            result.setProduceTimestamp(LocalDateTime.parse(array[1], DateTimeFormatter.ISO_LOCAL_DATE_TIME));
                            break;
                        case "processTimeStamp":
                            if(array.length>1 && !array[1].startsWith("nul")){
                                result.setProcessTimeStamp(LocalDateTime.parse(array[1], DateTimeFormatter.ISO_LOCAL_DATE_TIME));
                            }
                            break;
                    }
                });

        return result;
    }
}
